package com.gy.hadoop.hadoopRPC;

import javax.swing.*;
import java.nio.DoubleBuffer;
import java.util.LinkedList;

/**
 * 模拟管理元数据
 */
public class FsEdit {

    public static void main(String[] args) {
        FsEdit fs = new FsEdit();
        for (int i = 0; i < 1000; i++) {
            int finalI = i;
            new Thread(() -> {
                for (int i1 = 0; i1 < 100; i1++) {
                    fs.logEdit(String.format("线程%s开始写数据%s", finalI, i1));
                }
            }).start();

        }
    }

    long taxid = 0l;


    DoubleBuffer doubleBuffer = new DoubleBuffer();
    /**
     * 每个线程变量
     */
    ThreadLocal<Long> threaLocal = new ThreadLocal<Long>();

    /**
     * 后台是否在xiecipan
     */
    boolean isSyncRunning = false;

    /**
     * 正在同步磁盘的内存中的最大事务ID
     */
    long maxTaxid = 0L;

    boolean isWait = false;

    /**
     * 写日志   核心方法
     * 需要循序写入
     *
     * @param log
     */
    public void logEdit(String log) {
        synchronized (this) {
            taxid++;
            threaLocal.set(taxid);
            EditLog editLog = new EditLog(taxid, log);
            doubleBuffer.write(editLog);
        }


        logFlush();
    }

    private void logFlush() {
        synchronized (this) {
            /**
             * 当前线程的事务ID
             */
            Long localTaxid = threaLocal.get();
            if (localTaxid <= maxTaxid) return;

            if (isWait) return;
            isWait = true;
            while (isSyncRunning) {
                try {
                    this.wait(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            isWait = false;

        }
        doubleBuffer.exchange();
        if (doubleBuffer.syscBuffer.size() > 0)
            maxTaxid = doubleBuffer.getMaxTaxid();
        isSyncRunning = true;

        doubleBuffer.flush();

        synchronized (this) {
            isSyncRunning = false;
            this.notifyAll();
        }
    }


    /**
     * 元数据信息
     */
    public class EditLog {
        /**
         * 事务ID
         */
        public long taxid;
        /**
         * 日志内容
         */
        public String log;

        public EditLog(long taxid, String log) {
            this.taxid = taxid;
            this.log = log;
        }

        @Override
        public String toString() {
            return "EditLog{" +
                    "taxid=" + taxid +
                    ", log='" + log + '\'' +
                    '}';
        }
    }

    public class DoubleBuffer {
        /**
         * 写数据
         */
        LinkedList<EditLog> currentBuffer = new LinkedList<EditLog>();
        /**
         * 用来持久化的内存
         */
        LinkedList<EditLog> syscBuffer = new LinkedList<EditLog>();


        /**
         * 写元数据信息
         *
         * @param editLog
         */
        public void write(EditLog editLog) {
            currentBuffer.add(editLog);
        }

        /**
         * 把数据写入磁盘
         */
        public void flush() {
            for (Object editLog : syscBuffer) {
                System.out.println(editLog);
            }
            syscBuffer.clear();
        }

        /**
         * 交换一下内存
         */
        public void exchange() {
            LinkedList tmp = currentBuffer;
            currentBuffer = syscBuffer;
            syscBuffer = tmp;
        }

        /**
         * 正在同步的日志的最大事务ID
         *
         * @return
         */
        public Long getMaxTaxid() {
            return syscBuffer.getLast().taxid;
        }
    }


}
